package net.bw.realtime.tmall.dws.funcion1;

import com.alibaba.fastjson.JSONObject;
import net.bw.realtime.tmall.common.utils.JdbcUtil;
import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Map;

/*
 * @ Author：liuyawei
 * @ Date：2025-05-30
 */
public class CEPWindowFunction {

    public static void toCepTrade(DataStream<TransactionEventBean> timeStream) {

        // 定义异常模式，同一用户在短时间内多次大额支付
        Pattern<TransactionEventBean, ?> pattern = Pattern.<TransactionEventBean>begin("first").where(
                new SimpleCondition<TransactionEventBean>() {
                    @Override
                    public boolean filter(TransactionEventBean TransactionEventBean) throws Exception {

                        return "pay_success".equals(TransactionEventBean.getActionType())
                                && (TransactionEventBean.getPrice() * TransactionEventBean.getQuantity()) > 10000;
                    }
                }
        ).next("second").where(
                new SimpleCondition<TransactionEventBean>() {
                    @Override
                    public boolean filter(TransactionEventBean TransactionEventBean) throws Exception {
                        return "pay_success".equals(TransactionEventBean.getActionType())
                                && (TransactionEventBean.getPrice() * TransactionEventBean.getQuantity()) > 10000;
                    }
                }
        ).within(Time.minutes(1));

        KeyedStream<TransactionEventBean, String> keyedStream = timeStream.keyBy(TransactionEventBean::getUserId);

        SingleOutputStreamOperator<String> cep_trade = CEP.pattern(keyedStream, pattern)
                .select(new PatternSelectFunction<TransactionEventBean, String>() {
                    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                    @Override
                    public String select(Map<String, List<TransactionEventBean>> pattern) throws Exception {
                        TransactionEventBean first = pattern.get("first").get(0);
                        TransactionEventBean second = pattern.get("second").get(0);

                        String result = "用户：" + first.getUserId() + "在" + first.getTs() + "发生异常交易，金额：" + first.getPrice() * first.getQuantity() + "元" +
                                "，在" + second.getTs() + "又发生异常交易，金额：" + second.getPrice() * second.getQuantity() + "元";

//						System.out.println(result);
                        JSONObject cepjson = new JSONObject();
                        cepjson.put("userId", first.getUserId());
                        cepjson.put("first_ts", first.getTs());
                        cepjson.put("first_date", sdf.format(new Date(first.getTs())));
                        cepjson.put("first_price", first.getPrice() * first.getQuantity());
                        cepjson.put("second_ts", second.getTs());
                        cepjson.put("second_date", sdf.format(new Date(second.getTs())));
                        cepjson.put("second_price", second.getPrice() * second.getQuantity());


                        return cepjson.toString();
                    }
                })
                .name("cep-trade-choose")
                .uid("cep-trade-choose");

        //KafkaUtil.producerKafka(cep_trade,"tmall-trade-abnormal-cep");

        //cep_trade.print();

        JdbcUtil.sinkToClickhouseUpsertJson(
                cep_trade,
                new String[]{"first_date", "second_date", "userId", "first_price", "second_price", "first_ts", "second_ts"}
                ,
                "INSERT INTO jtp_tmall_log.dws_outlier_log(\n" +
                        "    first_date, second_date, userId, first_price, \n" +
                        "    second_price, first_ts, second_ts\n" +
                        ") VALUES (\n" +
                        "    ?, ?, ?, ?, ?, ?, ?\n" +
                        ")"
        );

    }
}
